/*
 *  src/clean/write_thread.c
 *
 *  Harry Wei <harryxiyou@gmail.com> 2011 (C)
 */

#include <glib.h>
#include <time.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/time.h>

#include "clean.h"
#include "address.h"
#include "hlfs.h"
#include "misc.h"
#include "logger.h"

/*
 * handle: move the obsolete segment's alive blocks into buf.
 * @para ctrl: the global control of hlfs.
 * @para su: the current su structure.
 * @para log_idx: the alive log index.
 * @para buf_size: the length of the buf.
 * @return: the buf address.
 */
char *handle(struct hlfs_ctrl *ctrl, struct segment_usage *su, uint64_t log_idx, uint64_t *buf_size)
{
	int len = 0;
	int j = 0;
	int i = 0;
	int tmp = 0;
	uint32_t cur_block_addr = 0;
	uint32_t block_addr = 0;
	uint64_t seg_size = 0;
	struct log_header *lh = NULL;
	struct inode *std_inode = NULL;
	struct inode inode;
	char *buf = NULL;
	char db_buf[(ctrl->sb).block_size];
	char *std_segname = NULL;
	char *segname = NULL;
	bs_file_t file = NULL;

	buf = (char *)g_malloc0(sizeof(char) * LOG_BUF);
	if (!buf) {
		g_message("%s - Allocate error!\n", __func__);
		exit(EXIT_FAILURE);
	}
	lh = (struct log_header *)g_malloc0(sizeof(struct log_header));
	if (!lh) {
		g_message("%s - Allocate error!\n", __func__);
		exit(EXIT_FAILURE);
	}
	printf("ctrl->last_segno %lld\n", ctrl->last_segno);
	printf("su->segno %lld\n", su->segno);
	std_segname = build_segfile_name(ctrl->last_segno - 1);
	std_inode = load_inode(ctrl, std_segname, get_last_inode_addr(ctrl, std_segname));
	segname = build_segfile_name(su->segno);
	printf("seg_file: %s\n", segname);
	if (NULL == (file = ctrl->storage->bs_file_open(ctrl->storage, segname, BS_READONLY))) { /* XXX if here with BS_WRITEABLE, bs_file_pread will give the len '-1' */
		g_message("%s - open file error!\n");
		exit(EXIT_FAILURE);
	}
	printf("log_idx %lld\n", log_idx);
	for (i = 0; i < log_idx; i++) {
		printf("seg_size %ld\n", seg_size);
		sleep(1);
		if ((uint32_t)LOG_HEADER_LENGTH != (len = ctrl->storage->bs_file_pread(ctrl->storage, file, lh, (uint32_t)LOG_HEADER_LENGTH, seg_size))) { /* XXX bug here, can not access */
			g_message("len -----------------> %d\n", len);
			g_message("%s - 1 pread error!\n", __func__);
			exit(EXIT_FAILURE);
		}
		seg_size += lh->log_size;
	}
		printf("seg_size %ld\n", seg_size);
		sleep(1);
	if (INODE_LEN != ctrl->storage->bs_file_pread(ctrl->storage, 
			file, &inode, INODE_LEN, seg_size - INODE_LEN - INODEM_LEN)) {
		g_message("%s - 2 pread error!\n", __func__);
		exit(EXIT_FAILURE);
	}
	g_message("start: %lld\n", lh->start_db_no);
	g_message("end: %ld\n", (uint64_t) lh->db_num - lh->start_db_no + 1);
	for (i = lh->start_db_no, j = 0; i < (uint64_t) lh->db_num - lh->start_db_no + 1; i++, j++) {
		if ((ctrl->sb).block_size != ctrl->storage->bs_file_pread(ctrl->storage, file, db_buf, (ctrl->sb).block_size, seg_size - lh->log_size + LOG_HEADER_LENGTH + j * (ctrl->sb).block_size)) {
			g_message("%s - pread error!\n", __func__);
			exit(EXIT_FAILURE);
		}
		g_message("%s --  1 pass here!!!\n", __func__);
		cur_block_addr = get_db_addr(ctrl, std_inode, i);
		block_addr = get_db_addr(ctrl, &inode, i);
		g_message("%s --  2 pass here!!!\n", __func__);
		printf("i-------------->%d\n", i);
		if ((cur_block_addr == block_addr)) {
			printf("tmp-------------->%d\n", tmp);
			memcpy(buf + *buf_size, db_buf, (ctrl->sb).block_size);
			tmp += 1;
			*buf_size += (ctrl->sb).block_size;
		}
	}
	printf("handle over!!!\n");
	sleep(5);
	g_free(lh);
	return (char *) buf;
}

/*
 * log_dump: dump the log's alive blocks into current segfile.
 * @para ctrl: the global control of hlfs.
 * @para buf: the contents of alive blocks.
 * @para buf_size: the length of the buf.
 * @return: void.
 */
void log_dump(struct hlfs_ctrl *ctrl, char *buf, uint64_t buf_size)
{
	char *segname = NULL;
	bs_file_t file = NULL;

	if (ctrl->last_offset > (ctrl->sb).seg_size) {
		ctrl->last_offset = 0;
		ctrl->last_segno += 1;
	}
	segname = build_segfile_name(ctrl->last_segno);
	if (0 == ctrl->last_offset) {
		if (NULL == (file = ctrl->storage->bs_file_create(ctrl->storage, segname))) {
			g_message("%s - create file error!\n");
			exit(EXIT_FAILURE);
		}
		if (buf_size != ctrl->storage->bs_file_append(ctrl->storage, file, buf, buf_size)) {
			g_message("%s - append error!\n");
			exit(EXIT_FAILURE);
		}
	} else {
		if (NULL == (file = ctrl->storage->bs_file_open(ctrl->storage, segname, BS_WRITEABLE))) {
			g_message("%s - open file error!\n");
			exit(EXIT_FAILURE);
		}	
		if (buf_size != ctrl->storage->bs_file_append(ctrl->storage, file, buf, buf_size)) {
			g_message("%s - append error!\n");
			exit(EXIT_FAILURE);
		}
	}
	g_free(buf);
	ctrl->storage->bs_file_close(ctrl->storage, file);
	return ;
}

/*
 * g_async_timeout_pop: after time seconds, pop the queue.
 * @para ctrl: the global control structure of hlfs.
 * @para time: the time seconds for blocking.
 * @return: return the pop 'data'.
 */
gpointer g_async_queue_timeout_pop(struct hlfs_ctrl *ctrl, uint32_t time)
{
	/* XXX When i remove the sleep, it will block forever, why ??? */
	sleep(time);
	w_req_t *w_req = g_async_queue_pop(ctrl->request_aqueue);
//	g_message("%s -- sleep time %ld\n", time);
	if (0 == w_req->size || NULL == w_req) {
		return NULL;
	}
	return w_req;
}

/*
 * api_write_rewrite: handle the api write and invoke write thread stuffs.
 * @para ctrl: the global control structure.
 * @para write_buf: the contents for writing.
 * @para write_len: the length for writing.
 * @para pos: the position to write.
 * @return: void.
 */
void api_write_rewrite(struct hlfs_ctrl *ctrl, char *write_buf, uint32_t write_len, uint64_t pos)
{
	w_req_t *w_req = NULL;
	w_res_t *w_res = NULL;

	if (!g_thread_supported())
		g_thread_init(NULL);
	ctrl->request_aqueue = g_async_queue_new();
	ctrl->response_aqueue = g_async_queue_new();
	w_req = (w_req_t *)g_malloc0(sizeof(w_req_t));
	if (!w_req) {
		g_message("%s - Allocate Error!\n", __func__);
		exit(EXIT_FAILURE);
	}
#if 0
	w_res = (w_res_t *)g_malloc0(sizeof(w_res_t));
	if (!w_res) {
		g_message("%s - Allocate Error!\n", __func__);
		g_free(w_req);
		exit(EXIT_FAILURE);
	}
#endif
#if 1
	w_req->buf = write_buf;
	w_req->size = write_len;
	w_req->pos = pos;
#endif
	g_async_queue_push(ctrl->request_aqueue, w_req);
	g_message("%s - before g_thread_create\n", __func__);
	g_thread_create(write_thread, ctrl, TRUE, NULL);
//	g_thread_create(su_check_task, ctrl, TRUE, NULL);
//	g_message("%s - after g_thread_create\n", __func__);
	/* XXX Here just for blocking ??? Pops data from the queue. This function blocks until data become available.*/
	sleep(1);
	w_res = g_async_queue_pop(ctrl->response_aqueue);
	g_message("%s - after g_async_queue_pop ......\n", __func__);
//	sleep(5);
	g_free(w_req);
	return ;
}

/*
 * write_thread: write thread for handling the api's write
 * request and cleaning's rewrite request.
 * @para ctrl: the fs's control structure.
 * @return: void.
 */
void write_thread(struct hlfs_ctrl *ctrl)
{
	uint32_t i = 0;
	char *buff = NULL;
	uint64_t buf_size = 0;
	char *segname = NULL;
	uint32_t time = 1;
	uint8_t thread_need_stop = 0;
	uint32_t seg_idx = 0;
	uint64_t seg_size = 0;
	uint64_t log_idx = 1;
	uint32_t log_inarray_idx = 0;
	uint64_t tmp = 0;
	w_req_t *w_req = NULL;
	w_res_t *w_res = NULL;
	gboolean start_clean = TRUE;
	struct segment_usage *su = NULL;
	struct log_header lh;
	uint64_t bit_map = BIT_CHECK;
	bs_file_t file = NULL;

	g_message("%s -- handle this thread\n", __func__);
#if 0
	su = (struct segment_usage *)g_malloc0(sizeof(struct segment_usage));
	if (!su) {
		g_message("%s - Allocate Error!\n");
		exit(EXIT_FAILURE);
	}
	/* run the su check task */
//	su_table = su_check_task(ctrl, RUN);
	w_req = (w_req_t *)g_malloc0(sizeof(w_req_t));
	if (!w_req) {
		g_message("%s - Allocate Error!\n", __func__);
		g_free(su);
		exit(EXIT_FAILURE);
	}
#endif
	w_res = (w_res_t *)g_malloc0(sizeof(w_res_t));
	if (!w_res) {
		g_message("%s - Allocate Error!\n", __func__);
		g_free(su);
		g_free(w_req);
		exit(EXIT_FAILURE);
	}
	while (!thread_need_stop) {
		/* XXX if nothing in async queue, it return NULL ??? */
		if (NULL != (w_req = g_async_queue_timeout_pop(ctrl, time))) {
			g_message("%s -- handle the write thread!!!\n", __func__);
//			sleep(1);
			uint32_t ret = hlfs_write(ctrl, w_req->buf, w_req->size, w_req->pos);
			if (ret == w_req->size) {
				printf("%s -- pass here 1\n", __func__);
				w_res->err_no = 0;
				w_res->size = ret;
			} else {
				w_res->err_no = -1;
				w_res->size = -1;
			}
			g_async_queue_push(ctrl->response_aqueue, w_res);
			thread_need_stop = 1;
		} else {
			do {
				printf("handle segment recycle..........\n");
				if (seg_idx == ctrl->su_table->len) {
					printf("clean over ......\n");
					start_clean = FALSE;
					dump_segment_usage_table(ctrl);
					w_res->err_no = 0;
					w_res->size = -1;
					g_async_queue_push(ctrl->response_aqueue, w_res);
					continue;
				}
				if (!start_clean)
				continue;
				if (0 == seg_idx) {
					if (ctrl->storage->bs_file_is_exist(ctrl->storage, "segment_usage")) {
						start_clean = FALSE;
						continue;
					}
				}
				printf("su->su_table->len: %d\n", ctrl->su_table->len);
				load_segment_usage(ctrl);
				su = &g_array_index(ctrl->su_table, struct segment_usage, seg_idx);
//				struct inode inode = ctrl->inode;
				if (0 == su->alive_blocks) { /* the segment's alive blocks is 0, remove it */
					segname = build_segfile_name(su->segno);
					ctrl->storage->bs_file_delete(ctrl->storage, segname);
					g_array_remove_index(ctrl->su_table, su->segno);
				} else if (su->alive_blocks < WATERLEVEL) { /* the seg's alive blocks is low */
					/* move the alive blocks into a new segment then remove this obsolete segment */
					segname = build_segfile_name(su->segno);
					bs_file_info_t *info = NULL;
					if (NULL == (info = ctrl->storage->bs_file_info(ctrl->storage, segname))) {
						g_message("%s -- get segfile info error!\n", __func__);
						exit(EXIT_FAILURE);
					}
#if 1
					if (NULL == (file = ctrl->storage->bs_file_open(ctrl->storage, segname, BS_READONLY))) {
						g_message("%s -- bs_file_open error!\n", __func__);
						exit(EXIT_FAILURE);
					}
#endif
					while (log_idx < 64llu * MAP_NUM && seg_size < info->size) {
#if 1
						if (LOG_HEADER_LENGTH != ctrl->storage->bs_file_pread(ctrl->storage, file, &lh, LOG_HEADER_LENGTH, seg_size)) {
							g_message("%s -- bs_file_pread error!\n", __func__);
							exit(EXIT_FAILURE);
						}
						seg_size += lh.log_size;
#endif
						if (bit_map &= su->alive_log_bitmap[tmp]) {
							buff = handle(ctrl, su, log_idx, &buf_size);
							g_message("%s -- dump------------->\n", __func__);
							sleep(5);
							log_dump(ctrl, buff, buf_size);
							g_message("%s -- after dump------------->\n", __func__);
							sleep(5);
						}
						bit_map <<= 1;
						if (0 == (log_idx % 64)) {
							tmp += 1;
							bit_map = BIT_CHECK;
						}
						log_idx += 1;
						g_message("%s -- log_idx ------------------> %d\n", __func__, log_idx);
					}
					ctrl->storage->bs_file_delete(ctrl->storage, segname);
					g_array_remove_index(ctrl->su_table, su->segno);
					ctrl->storage->bs_file_close(ctrl->storage, file);
				}
				tmp = 0;
				log_idx = 1;
				seg_size = 0;
				bit_map = BIT_CHECK;
				printf("seg_idx is %d --------------->here!\n", seg_idx);
				seg_idx += 1;
			} while (start_clean);
		}
	}
}
